from ask_ai import *
import time
import re
import pandas as pd
import redis
import pickle
import logging
import configparser
from sqlalchemy import create_engine
import pymysql
import os
import sys
import datetime
import pyarrow as pa
import pyarrow.parquet as pq
from multiprocessing import Pool
import numpy as np
from collections import defaultdict
from clickhouse_driver import Client
import concurrent.futures
import requests


client = Client('192.168.8.100', send_receive_timeout=7200, port=9000, user='default', password='dgls', database='stock')

# 连接数据库 (假设你使用的是ClickHouse数据库，你可以根据实际情况更改连接方式)
engine = create_engine('clickhouse+http://default:dgls@192.168.8.100:8123/stock')

# 查询数据
query = """
    select * from stock.stock_fupan where date='20240731' and stock_code  NOT IN (SELECT stock_code FROM stock.stock_fupan_gn) and stock_name not like '%%ST%%'
    and stock_name not like '中国%%' and stock_name not like '%%银行%%' and stock_name not like '%%证券%%' and stock_name not like '%%地产%%' and industry not like '%%白酒%%'
    and stock_code not like '8%%'
"""

df = pd.read_sql(query, engine)

# 定义处理每一行数据的函数
def process_row(row):
    row_dict = row.to_dict()
    question = f"通过联网搜索，最近一年{row_dict['stock_name']}在资本市场上的最主要的炒作概念是啥，如果网上找不到，或者没有相关信息，或者内容特别少，直接回复找不到或者回复信息太少。"
    print(question)
    row_dict["remark"] = askKiMiByWeb(question)
    print(row_dict["remark"])

    client.execute(f"DELETE from stock.stock_fupan_gn where stock_name='{row_dict['stock_name']}'")
    # 准备插入数据
    columns = ','.join(row_dict.keys())
    values = tuple(row_dict.values())
    client.execute(f'INSERT INTO stock.stock_fupan_gn ({columns}) VALUES', [values])

# 使用 ThreadPoolExecutor 进行多线程处理
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
    # 将 DataFrame 的每一行提交给线程池
    futures = [executor.submit(process_row, row) for index, row in df.iterrows()]

    # 等待所有线程完成
    for future in concurrent.futures.as_completed(futures):
        try:
            future.result()  # 如果有异常，在这里会抛出
        except Exception as e:
            print(f"处理时发生错误: {e}")

print("所有数据已成功处理并插入到 ClickHouse 数据库")


